package net.chery.service;

import jakarta.annotation.PostConstruct;
import jakarta.persistence.PostLoad;
import lombok.extern.slf4j.Slf4j;
import net.chery.Cons;
import net.chery.entity.DataType;
import net.chery.entity.VehProduce;
import net.chery.exec.BatteryExec;
import net.chery.exec.HiveExec;
import net.chery.repository.VehProduceRepository;
import net.chery.rm.CopyProperties;
import net.chery.rm.DmsRowMapper;
import net.chery.rm.MesRowMapper;
import net.chery.rm.SimplePropertyRowMapper;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @author gai
 * @description etl
 * @date 2024-04-17
 */
@Slf4j
@Service
public class EtlService implements InitializingBean {
    private final VehProduceRepository vehProduceRepository;
    private final HiveExec hiveExec;
    private final BatteryExec batteryExec;
    private final SaveService saveService;
    private final ApplicationEventPublisher eventPublisher;

    public EtlService(VehProduceRepository vehProduceRepository, HiveExec hiveExec, BatteryExec batteryExec, SaveService saveService, ApplicationEventPublisher eventPublisher) {
        this.vehProduceRepository = vehProduceRepository;
        this.hiveExec = hiveExec;
        this.batteryExec = batteryExec;
        this.saveService = saveService;
        this.eventPublisher = eventPublisher;
    }

    @Override
    public void afterPropertiesSet() {
    }

    private static long getToday() {
        LocalDate date = LocalDate.now();
        return Date.from(date.atStartOfDay().atZone(ZoneId.systemDefault()).toInstant()).getTime();
    }

    @Scheduled(initialDelay = 10000)
    public void start(){
       syncProduce();
    }

    /**
     * 同步生产信息
     */
    @Scheduled(cron = "0 0 3 * * ?")
    public void syncProduce() {
        long today = getToday();
        final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        Date date = new Date(today + TimeUnit.DAYS.toMillis(-7));
        String batteryDate = sdf.format(date);
        String dmsDate = sdf.format(date);
        Map<String, VehProduce> vehProduceMap = Cons.mapObject(vehProduceRepository
                        .findByCreateTimeGreaterThan(date),
//                        .findAll(),
                VehProduce::getVin);
        CopyProperties<VehProduce> produceCopy = getProduceCopy();
        CopyProperties<VehProduce> saleCopy = getSaleCopy();
        Consumer<VehProduce> function = getFunction(produceCopy, saleCopy);
        Function<String, VehProduce> formCache = vehProduceRepository.getFormCache(vehProduceMap);
        log.info("start findUnSyncData");
        findUnSyncData(vehProduceMap,date,produceCopy,saleCopy);
        log.info("start battery produce");
        batteryExec.execEntity("select vin,\n" +
                        "       create_time produce_create_time,\n" +
                        "       veh_model_name,\n" +
                        "       is_match,\n" +
                        "       vehicle_name,\n" +
                        "       offline_production_time,\n" +
                        "       vehicle_brand,\n" +
                        "       if(report_time is null or report_time = '', null, str_to_date(report_time,'%Y-%m-%d %H:%i:%s')) report_time,\n" +
                        "       sync_state,\n" +
                        "       sync_state_sh from sys_veh_produce where create_time>?",
//                        "       sync_state_sh from sys_veh_produce",
                new SimplePropertyRowMapper<>(VehProduce.class), vehProduce -> {
                    VehProduce cache = formCache.apply(vehProduce.getVin());
                    cache.addType(DataType.PRODUCE);
                    produceCopy.copy(cache, vehProduce);
                }, batteryDate);
//                });
        log.info("start battery sale");
        batteryExec.execEntity("select vin,\n" +
                        "create_time sale_create_time,\n" +
                        "sale_time,\n" +
                        "sale_area,\n" +
                        "license_plate,\n" +
                        "license_additional_time,\n" +
                        "sale_unit_code,\n" +
                        "sale_unit_name,\n" +
                        "if(report_time is null or report_time = '', null, str_to_date(report_time,'%Y-%m-%d %H:%i:%s')) sale_report_time,\n" +
                        "sync_state sale_sync_state,\n" +
                        "sync_state_sh sale_sync_state_sh,\n" +
                        "sale_address,\n" +
                        "sale_area_sh from sys_veh_sale where create_time>?",
//                        "sale_area_sh from sys_veh_sale",
                new SimplePropertyRowMapper<>(VehProduce.class), vehProduce -> {
                    VehProduce cache = formCache.apply(vehProduce.getVin());
                    cache.addType(DataType.SALE);
                    saleCopy.copy(cache, vehProduce);
                }
                , batteryDate);
//                );
        Set<String> vehModels = getVehModels();

        dealMes(vehModels, function, formCache, date);
        dealDms(vehModels, function, formCache, dmsDate);
        log.info("start saveFile");
        String filename = saveService.write2File(vehProduceMap.values());
        log.info("start save");
        eventPublisher.publishEvent(new FileEvent(filename,date));
        log.info("end syncSale");
    }

    private void findUnSyncData(Map<String, VehProduce> vehProduceMap,
                                Date date,
                                CopyProperties<VehProduce> produceCopy,
                                CopyProperties<VehProduce> saleCopy){
        List<VehProduce> vehProduces=vehProduceRepository.findUnSyncData(date);
        for (VehProduce produce:vehProduces){
            if(DataType.PRODUCE.has(produce.getType())){
                boolean needSh=Cons.YES.equals(produce.getNeedSh())&&Cons.NO.equals(produce.getSyncStateSh());
                if(Cons.NO.equals(produce.getSyncState())||needSh){
                    batteryExec.execEntity("select vin,\n" +
                                    "       create_time produce_create_time,\n" +
                                    "       veh_model_name,\n" +
                                    "       is_match,\n" +
                                    "       vehicle_name,\n" +
                                    "       offline_production_time,\n" +
                                    "       vehicle_brand,\n" +
                                    "       if(report_time is null or report_time = '', null, str_to_date(report_time,'%Y-%m-%d %H:%i:%s')) report_time,\n" +
                                    "       sync_state,\n" +
                                    "       sync_state_sh from sys_veh_produce where vin = ?",
                            new SimplePropertyRowMapper<>(VehProduce.class), vehProduce -> {
                                produce.addType(DataType.PRODUCE);
                                produceCopy.copy(produce, vehProduce);
                            }, produce.getVin());
                }
            }
            if(DataType.SALE.has(produce.getType())){
                boolean needSh=Cons.YES.equals(produce.getNeedSh())&&Cons.NO.equals(produce.getSaleSyncStateSh());
                if(Cons.NO.equals(produce.getSaleSyncState())||needSh){
                    batteryExec.execEntity("select vin,\n" +
                                    "create_time sale_create_time,\n" +
                                    "sale_time,\n" +
                                    "sale_area,\n" +
                                    "license_plate,\n" +
                                    "license_additional_time,\n" +
                                    "sale_unit_code,\n" +
                                    "sale_unit_name,\n" +
                                    "if(report_time is null or report_time = '', null, str_to_date(report_time,'%Y-%m-%d %H:%i:%s')) sale_report_time,\n" +
                                    "sync_state sale_sync_state,\n" +
                                    "sync_state_sh sale_sync_state_sh,\n" +
                                    "sale_address,\n" +
                                    "sale_area_sh from sys_veh_sale where vin = ?",
                            new SimplePropertyRowMapper<>(VehProduce.class), vehProduce -> {
                                produce.addType(DataType.SALE);
                                saleCopy.copy(produce, vehProduce);
                            }, produce.getVin());
                }
            }
            vehProduceMap.put(produce.getVin(),produce);
        }
    }

    /**
     * 对于从hive的数据进行信息补充
     */
    private Consumer<VehProduce> getFunction(CopyProperties<VehProduce> produceCopy,
                                             CopyProperties<VehProduce> saleCopy) {
        return (produce) -> {
            if (!DataType.PRODUCE.has(produce.getType())) {
                batteryExec.execEntity("select vin,\n" +
                                "       create_time produce_create_time,\n" +
                                "       veh_model_name,\n" +
                                "       is_match,\n" +
                                "       vehicle_name,\n" +
                                "       if(report_time is null or report_time = '', null, str_to_date(report_time,'%Y-%m-%d %H:%i:%s')) report_time,\n" +
                                "       offline_production_time,\n" +
                                "       vehicle_brand,\n" +
                                "       sync_state,\n" +
                                "       sync_state_sh from sys_veh_produce where vin = ?",
                        new SimplePropertyRowMapper<>(VehProduce.class), vehProduce -> {
                            produceCopy.copy(produce, vehProduce);
                            produce.addType(DataType.PRODUCE);
                        }, produce.getVin());
            }
            if (!DataType.SALE.has(produce.getType())) {
                batteryExec.execEntity("select vin,\n" +
                                "create_time sale_create_time,\n" +
                                "sale_time,\n" +
                                "sale_area,\n" +
                                "license_plate,\n" +
                                "if(report_time is null or report_time = '', null, str_to_date(report_time,'%Y-%m-%d %H:%i:%s')) sale_report_time,\n" +
                                "license_additional_time,\n" +
                                "sale_unit_code,\n" +
                                "sale_unit_name,\n" +
                                "sync_state sale_sync_state,\n" +
                                "sync_state_sh sale_sync_state_sh,\n" +
                                "sale_address,\n" +
                                "sale_area_sh from sys_veh_sale where vin = ?",
                        new SimplePropertyRowMapper<>(VehProduce.class), vehProduce -> {
                            saleCopy.copy(produce, vehProduce);
                            produce.addType(DataType.SALE);
                        }, produce.getVin());
            }
        };
    }

    /**
     * produce数据复制
     */
    public CopyProperties<VehProduce> getProduceCopy() {
        return (cache, vehProduce) -> {
            cache.setVehModelName(vehProduce.getVehModelName());
            cache.setProduceCreateTime(vehProduce.getProduceCreateTime());
            cache.setVehicleName(vehProduce.getVehicleName());
            cache.setIsMatch(vehProduce.getIsMatch());
            cache.setOfflineProductionTime(vehProduce.getOfflineProductionTime());
            cache.setVehicleBrand(vehProduce.getVehicleBrand());
            cache.setSyncState(vehProduce.getSyncState());
            if(vehProduce.getReportTime()!=null){
                cache.setReportTime(vehProduce.getReportTime());
            }
            cache.setSyncStateSh(vehProduce.getSyncStateSh());
        };
    }

    /**
     * sale数据复制
     */
    public CopyProperties<VehProduce> getSaleCopy() {
        return (cache, vehProduce) -> {
            cache.setSaleCreateTime(vehProduce.getSaleCreateTime());
            cache.setSaleTime(vehProduce.getSaleTime());
            cache.setSaleArea(vehProduce.getSaleArea());
            cache.setSaleSyncState(vehProduce.getSaleSyncState());
            if(vehProduce.getSaleReportTime()!=null){
                cache.setSaleReportTime(vehProduce.getSaleReportTime());
            }
            cache.setLicensePlate(vehProduce.getLicensePlate());
            cache.setLicenseAdditionalTime(vehProduce.getLicenseAdditionalTime());
            cache.setSaleUnitCode(vehProduce.getSaleUnitCode());
            cache.setSaleUnitName(vehProduce.getSaleUnitName());
            cache.setSaleSyncStateSh(vehProduce.getSaleSyncStateSh());
            cache.setSaleAddress(vehProduce.getSaleAddress());
            cache.setSaleAreaSh(vehProduce.getSaleAreaSh());
        };
    }

    /**
     * 处理mes数据
     */
    private void dealMes(Set<String> vehModels, Consumer<VehProduce> function, Function<String, VehProduce> formCache, Object... params) {
        MesRowMapper mesRowMapper = new MesRowMapper(vehModels, function, formCache);
        String sqlFormat = "select mes.clxh," +
                " mes.is_print," +
                " mes.clmc," +
                " mes.clpp," +
                " mes.rlzl," +
                " mes.vin," +
                " mes.print_time," +
                " mes.gmt_modified," +
                " mes.is_deleted\n" +
                "from %s mes\n" +
                "where is_deleted = 0\n" +
//                "  and is_print = 1";
                "  and is_print = 1\n" +
                "  and gmt_modified > ?";
        log.info("start mes cy");
        hiveExec.execEntity(String.format(sqlFormat,
                "ods_qr.ods_qr_mes_cy_tm_bas_checkout_record_df"
        ), mesRowMapper, (map -> {
            if (map != null) {
                map.setFactory("超一厂");
            }
        }), params);
//        }));
        log.info("start mes qd");
        hiveExec.execEntity(String.format(sqlFormat,
                "ods_qr.ods_qr_mes_qd_tm_bas_checkout_record_df"
        ), mesRowMapper, (map -> {
            if (map != null) {
                map.setFactory("青岛");
            }
        }), params);
//        }));
    }

    /**
     * 处理dms数据
     */
    private void dealDms(Set<String> vehModels, Consumer<VehProduce> function, Function<String, VehProduce> formCache, Object... params) {
        String sqlFormat = "select modifytime,\n" +
                "       createtime,\n" +
                "       branchcode,\n" +
                "       branchname,\n" +
                "       annoucementnumber,\n" +
                "       manufacturername,\n" +
                "       vin,\n" +
                "       licenseplate,\n" +
                "       registrationtime,\n" +
                "       productcode,\n" +
                "       productname,\n" +
                "       %s saledate,\n" +
                "       salesdealercode,\n" +
                "       salesdealername\n" +
//                "from %s\n";
                "from %s\n" +
                "where modifytime > ?";
        DmsRowMapper dmsRowMapper = new DmsRowMapper(vehModels, function, formCache);
        log.info("start dms qr");
        hiveExec.execEntity(String.format(sqlFormat,"filingtime", "ods_qr.ods_qr_dms_vehiclesold_di"), dmsRowMapper, (map -> {
        }), params);
//        }));
        log.info("start dms xt");
        hiveExec.execEntity(String.format(sqlFormat,"", "ods_xt.ods_xt_dms_vehiclesold_di"), dmsRowMapper, (map -> {
        }), params);
//        }));
    }

    /**
     * 获取电车的车辆型号
     */
    private Set<String> getVehModels() {
        Set<String> res = new HashSet<>();
        batteryExec.exec("select veh_model_id from sys_veh_model", resultSet -> {
            try {
                res.add(resultSet.getString(1));
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        });
        return res;
    }

    public List<List<String>> desc(String table) {
        return hiveExec.desc(table);
    }

    public List<List<Object>> query(String sql) {
        List<List<Object>> list = new ArrayList<>();
        List<Object> cols = new ArrayList<>();
        hiveExec.exec(sql, rs -> {
            try {
                if (cols.isEmpty()) {
                    ResultSetMetaData metaData = rs.getMetaData();
                    int len = metaData.getColumnCount();
                    for (int i = 1; i <= len; i++) {
                        cols.add(metaData.getColumnName(i));
                    }
                    list.add(cols);
                }
                List<Object> data = new ArrayList<>();
                for (int i = 1; i <= cols.size(); i++) {
                    data.add(rs.getObject(i));
                }
                list.add(data);
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }
        });

        return list;
    }
}
